package org.apache.nutch.scoring.webgraph.algorithms;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import org.apache.hadoop.io.ObjectWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableUtils;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.util.ToolRunner;
import org.apache.nutch.scoring.webgraph.LinkDatum;
import org.apache.nutch.scoring.webgraph.LinkRank;
import org.apache.nutch.scoring.webgraph.Node;
import org.apache.nutch.util.NutchConfiguration;

public class Hits extends LinkRank
{
	/**
	 * Runs a single link analysis iteration.
	 */
	private static class Analyzer implements
			Mapper<Text, Writable, Text, ObjectWritable>,
			Reducer<Text, ObjectWritable, Text, Node>
	{
		private JobConf conf;

		public void configure(JobConf conf)
		{
			this.conf = conf;

      System.err.println("");
      System.err.println("ANALYZA");
		}

		/**
		 * Convert values to ObjectWritable
		 */
		public void map(Text key, Writable value,
				OutputCollector<Text, ObjectWritable> output, Reporter reporter)
				throws IOException
		{
			ObjectWritable objWrite = new ObjectWritable();
			objWrite.set(WritableUtils.clone(value, conf));
			output.collect(key, objWrite);
		}

		/**
		 * Indegree reduce-function.
		 * 
		 */
		public void reduce(Text key, Iterator<ObjectWritable> values,
				OutputCollector<Text, Node> output, Reporter reporter)
				throws IOException
		{
			Node node = null;
			float authorityScore = 0.0f;
			float hubScore = 0.0f;

			while (values.hasNext())
			{
				Object value = values.next().get();
				
				if (value instanceof Node)
					node = (Node) value;
				else if (value instanceof LinkDatum)
				{
					LinkDatum link = (LinkDatum)value;
					
					if (link.getLinkType() == LinkDatum.INLINK)
						authorityScore += link.getScore();
					else
						hubScore += link.getScore();
				}
			}
			
			// store the score in a temporary NodeDb
			Node outNode = (Node) WritableUtils.clone(node, conf);
			LOG.info("Setting new score for " + key + ": authority score = " +
			  authorityScore + ", hib score = " + hubScore);

			outNode.setInlinkScore(authorityScore);
			outNode.setOutlinkScore(hubScore);
			output.collect(key, outNode);

      System.err.println("ANALYZER: New weights for " + key + ": " +
      		"(" + node.getInlinkScore() + ", " + node.getOutlinkScore() + ") -> " +
      		"(" + outNode.getInlinkScore() + ", " + outNode.getOutlinkScore() + ")");

		}

		public void close() throws IOException
		{
		}
	}

	public int run(String[] args) throws Exception
	{
		setAnalyzerMapClass(Analyzer.class);
		setAnalyzerReduceClass(Analyzer.class);
		// Using default Normalizer and Converger

		getConf().setInt("link.analyze.num.iterations", Integer.MAX_VALUE);

		return super.run(args);
	}

	/**
	 * @param args
	 * @throws Exception 
	 */
	public static void main(String[] args) throws Exception
	{
		LinkRank alg = new Hits();

		int res = ToolRunner.run(NutchConfiguration.create(), alg, args);
		System.exit(res);
	}

}
